import pymysql
import logging
from datetime import datetime
import time
from tqdm import tqdm
import sys
import psutil
import gc
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
import threading
from itertools import chain
import traceback

# 配置参数
BATCH_SIZE = 2000
THREAD_POOL_SIZE = 16
MAX_RETRIES = 3
RETRY_WAIT_TIME = 5
QUEUE_SIZE = 40

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'goods_sync_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# 数据库连接配置
ecs_config = {
    'host': 'localhost',
    'user': 'root',
    'password': '123456',
    'db': 'zazhi2007',
    'charset': 'utf8',
    'port': 3306
}

fa_config = {
    'host': '211.154.25.60',
    'user': 'zazhi4',
    'password': 'DBdz3nr8NeAxY37S',
    'db': 'zazhi4',
    'charset': 'utf8mb4',
    'port': 3307
}

class DatabaseSync:
    def __init__(self):
        self.goods_type_map = {}
        self.sync_stats = {
            'total': 0,
            'success': 0,
            'failed': 0,
            'skipped': 0
        }
        self.performance_stats = {
            'start_time': None,
            'end_time': None,
            'memory_usage': []
        }
        self.stats_lock = threading.Lock()
        self.data_queue = Queue(maxsize=QUEUE_SIZE)
        self.progress_bar = None
        self.thread_local = threading.local()

    def get_ecs_connection(self):
        """获取ECS数据库连接"""
        try:
            if not hasattr(self.thread_local, 'ecs_conn'):
                self.thread_local.ecs_conn = pymysql.connect(**ecs_config)
            return self.thread_local.ecs_conn
        except Exception as e:
            logger.error(f"获取ECS数据库连接失败: {str(e)}")
            raise

    def get_fa_connection(self):
        """获取FA数据库连接"""
        try:
            if not hasattr(self.thread_local, 'fa_conn'):
                self.thread_local.fa_conn = pymysql.connect(**fa_config)
            return self.thread_local.fa_conn
        except Exception as e:
            logger.error(f"获取FA数据库连接失败: {str(e)}")
            raise

    def get_goods_type_map(self):
        """获取商品类型映射"""
        try:
            conn = self.get_ecs_connection()
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            cursor.execute("SELECT cat_id, cat_name FROM ecs_goods_type")
            self.goods_type_map = {row['cat_id']: row['cat_name'] for row in cursor.fetchall()}
            cursor.close()
            logger.info(f"获取到 {len(self.goods_type_map)} 个商品类型映射")
        except Exception as e:
            logger.error(f"获取商品类型映射失败: {str(e)}")
            raise

    def producer(self):
        """数据生产者"""
        try:
            ecs_conn = self.get_ecs_connection()
            ecs_cursor = ecs_conn.cursor(pymysql.cursors.SSDictCursor)
            
            # 先查询特定商品作为测试
            test_query = """
                SELECT 
                    goods_name,
                    goods_type,
                    is_on_sale,
                    CAST(is_on_sale AS SIGNED) as is_on_sale_int
                FROM ecs_goods 
                WHERE goods_name = '中国典籍与文化'
            """
            ecs_cursor.execute(test_query)
            test_record = ecs_cursor.fetchone()
            if test_record:
                logger.info(f"测试商品数据: {test_record}")
            
            offset = 0
            while True:
                ecs_cursor.execute("""
                    SELECT 
                        goods_name,
                        goods_type,
                        is_on_sale,
                        CAST(is_on_sale AS SIGNED) as is_on_sale_int
                    FROM ecs_goods 
                    LIMIT %s, %s
                """, (offset, BATCH_SIZE))
                
                records = ecs_cursor.fetchall()
                if not records:
                    break
                
                # 记录每条数据的详细信息
                for record in records:
                    logger.debug(f"原始数据: {record}")
                    
                batch_data = []
                for row in records:
                    if row['goods_name']:
                        # 确保is_on_sale是整数
                        row['is_on_sale'] = int(row['is_on_sale_int'])
                        batch_data.append(row)
                
                if batch_data:
                    self.data_queue.put(batch_data)
                
                offset += BATCH_SIZE
                
        except Exception as e:
            logger.error(f"生产者错误: {str(e)}")
            logger.error(f"错误堆栈: {traceback.format_exc()}")
            raise
        finally:
            for _ in range(THREAD_POOL_SIZE):
                self.data_queue.put(None)

    def update_batch(self, batch_data):
        """批量更新数据"""
        fa_conn = self.get_fa_connection()
        fa_cursor = fa_conn.cursor()
        success_count = 0
        try:
            update_sql = """
                UPDATE fa_shop_goods 
                SET 
                    goods_type = %s,
                    goods_type_detail = %s,
                    status = %s,
                    updatetime = %s
                WHERE title = %s
            """
            
            for data in batch_data:
                try:
                    goods_name = data['goods_name']
                    goods_type = data['goods_type']
                    is_on_sale = data['is_on_sale']
                    
                    goods_type_detail = self.goods_type_map.get(goods_type, '')
                    status = 'normal' if is_on_sale == 1 else 'hidden'
                    current_time = int(time.time())

                    logger.info(f"""
                    更新商品:
                    - 名称: {goods_name}
                    - is_on_sale: {is_on_sale}
                    - status: {status}
                    - type: {goods_type}
                    - detail: {goods_type_detail}
                    """)

                    fa_cursor.execute(update_sql, (
                        goods_type,
                        goods_type_detail,
                        status,
                        current_time,
                        goods_name
                    ))
                    success_count += 1

                except Exception as e:
                    logger.error(f"更新单条记录失败: goods_name={goods_name}, error={str(e)}")
                    logger.error(f"错误数据: {data}")
                    continue

            fa_conn.commit()
            
            with self.stats_lock:
                self.sync_stats['success'] += success_count
                if self.progress_bar:
                    self.progress_bar.update(len(batch_data))
            
        except Exception as e:
            fa_conn.rollback()
            logger.error(f"批量更新失败: {str(e)}")
            logger.error(f"错误堆栈: {traceback.format_exc()}")
            with self.stats_lock:
                self.sync_stats['failed'] += len(batch_data)
            raise
        finally:
            fa_cursor.close()

    def consumer(self):
        """数据消费者"""
        while True:
            try:
                batch_data = self.data_queue.get()
                if batch_data is None:
                    self.data_queue.task_done()
                    break
                    
                retry_count = 0
                while retry_count < MAX_RETRIES:
                    try:
                        self.update_batch(batch_data)
                        break
                    except Exception as e:
                        retry_count += 1
                        if retry_count == MAX_RETRIES:
                            logger.error(f"更新失败，已达到最大重试次数: {str(e)}")
                            break
                        logger.warning(f"更新失败，第 {retry_count} 次重试")
                        time.sleep(RETRY_WAIT_TIME)
                
                self.data_queue.task_done()
                
            except Exception as e:
                logger.error(f"消费者错误: {str(e)}")
                logger.error(f"错误堆栈: {traceback.format_exc()}")
                continue

    def sync_goods_data(self):
        """同步商品数据"""
        try:
            self.performance_stats['start_time'] = time.time()
            
            # 获取商品类型映射
            self.get_goods_type_map()
            
            # 获取总记录数
            ecs_conn = self.get_ecs_connection()
            cursor = ecs_conn.cursor()
            cursor.execute("SELECT COUNT(*) as total FROM ecs_goods")
            total_records = cursor.fetchone()[0]
            
            if total_records == 0:
                logger.error("ECS商品表没有数据")
                return
                
            self.sync_stats['total'] = total_records
            cursor.close()
            
            # 创建进度条
            self.progress_bar = tqdm(total=total_records, desc="同步进度")
            
            # 创建线程池
            with ThreadPoolExecutor(max_workers=THREAD_POOL_SIZE) as executor:
                # 启动生产者
                producer_future = executor.submit(self.producer)
                
                # 启动消费者
                consumer_futures = [
                    executor.submit(self.consumer)
                    for _ in range(THREAD_POOL_SIZE)
                ]
                
                # 等待所有任务完成
                producer_future.result()
                for future in consumer_futures:
                    future.result()
            
            self.performance_stats['end_time'] = time.time()
            duration = self.performance_stats['end_time'] - self.performance_stats['start_time']
            
            logger.info(f"同步完成统计：")
            logger.info(f"总数={self.sync_stats['total']}, "
                       f"成功={self.sync_stats['success']}, "
                       f"失败={self.sync_stats['failed']}, "
                       f"跳过={self.sync_stats['skipped']}")
            logger.info(f"执行时间: {duration:.2f} 秒")
            
        except Exception as e:
            logger.error(f"同步过程出错: {str(e)}")
            logger.error(f"错误堆栈: {traceback.format_exc()}")
            raise
        finally:
            if hasattr(self, 'progress_bar') and self.progress_bar:
                self.progress_bar.close()

def main():
    sync = DatabaseSync()
    try:
        sync.sync_goods_data()
    except Exception as e:
        logger.error(f"同步失败: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()
